package main

import (
    "sync"
    "time"
    "github.com/gin-gonic/gin"
    "fmt"
    "encoding/json"
    "io/ioutil"
    "os"
    "path/filepath"
)

type Controller struct {
    stores          StoreSet
    tablesDefs      map[string]map[string]TableDefine
    mu              sync.RWMutex
    interval        int
    stopping        chan interface{}
    stopped         chan interface{}
    o, r, w         int64
    tableDefineFile string
    storeBasePath   string
}

func NewController(defpath, storepath string, interval int) (*Controller, error) {
    c := &Controller{
        stopping:        make(chan interface{}),
        stopped:         make(chan interface{}),
        tableDefineFile: defpath,
        storeBasePath:   storepath,
    }
    if err := c.LoadTableDef(); err != nil {
        return nil, err
    }
    if err := c.LoadLocalData(); err != nil {
        return nil, err
    }
    go c.CheckAndClose()
    return c, nil
}

func (c *Controller)Close(){
    close(c.stopping)
    <-c.stopped
}

func (c *Controller) CloseStore(closeAll bool){
    c.mu.Lock()
    defer c.mu.Unlock()
    ref:=c.stores
    if !closeAll{
        ref = c.stores.FilterByActive(c.o, c.r, c.w)
    }
    for _, s := range ref {
                if s.IsActive() {
                    s.Close()
                }
            }
}

func (c *Controller) CheckAndClose() {
    tick := time.Tick(time.Second * time.Duration(c.interval))
    for {
        select {
        case <-tick:
            c.CloseStore(false)
        case <-c.stopping:
            c.CloseStore(true)
            close(c.stopped)
        }
    }
}

func (c *Controller) LoadTableDef() error {
    c.mu.Lock()
    defer c.mu.Unlock()
    b, err := ioutil.ReadFile(c.tableDefineFile)
    if err != nil {
        return err
    }
    var defs []TableDefine
    if err := json.Unmarshal(b, defs); err != nil {
        return err
    }
    c.tablesDefs = make(map[string]map[string]TableDefine)
    for _, df := range defs {
        if _, ok := c.tablesDefs[df.Appid]; !ok {
            c.tablesDefs[df.Appid] = make(map[string]TableDefine)
        }
        if _, ok := c.tablesDefs[df.Appid][df.TableName]; ok {
            return fmt.Errorf("duplicated table name %s", df.TableName)
        }
        c.tablesDefs[df.Appid][df.TableName] = df
    }
    return nil
}

func (c *Controller) WalkLoad(path string, info os.FileInfo, err error) error {
    if !info.IsDir() {
        return nil
    }
    p, err := filepath.Rel(c.storeBasePath, path)
    if err != nil {
        return err
    }
    subdir, partition := filepath.Split(p)
    appid, tablename := filepath.Split(subdir)
    if appid == "" || tablename == "" || partition == "" {
        return nil
    }
    c.mu.Lock()
    defer c.mu.Unlock()
    if _, ok := c.tablesDefs[appid]; !ok {
        return nil
    }
    if _, ok := c.tablesDefs[appid][tablename]; !ok {
        return nil
    }
    par, err := NewPartition(partition)
    if err != nil {
        return nil
    }
    def := c.tablesDefs[appid][tablename]
    if def.DtLength != len(par.dt) {
        return fmt.Errorf("dt length not match %s", path)
    }
    store, err := NewRocksdbStore(def, *par)
    if err != nil {
        return err
    }
    c.stores = append(c.stores, store)
    return nil
}

func (c *Controller) LoadLocalData() error {
    return filepath.Walk(c.storeBasePath, c.WalkLoad)
}

func (c *Controller) createTableHandler(ctx *gin.Context) {
    ctx.AbortWithError(403, fmt.Errorf("not supported yet"))
}

func (c *Controller) getTableHandler(ctx *gin.Context) {
    def, err := c.getDefine(ctx)
    if err != nil {
        ctx.AbortWithError(404, err)
        return
    }
    b, err := json.Marshal(def)
    if err != nil {
        ctx.AbortWithError(500, err)
        return
    }
    ctx.JSON(200, b)
}

func (c *Controller) getDefine(ctx *gin.Context) (*TableDefine, error) {
    c.mu.RLock()
    defer c.mu.RUnlock()
    appid := ctx.Param("appid")
    tablename := ctx.Param("tablename")
    m, ok := c.tablesDefs[appid]
    if !ok {
        return nil, fmt.Errorf("appid %s not found", appid)
    }
    def, ok := m[tablename]
    if !ok {
        return nil, fmt.Errorf("tablename %s not found", tablename)
    }
    return &def, nil
}

func (c *Controller) writeHandler(ctx *gin.Context) {
    t := time.Now().UnixNano() / 1000 / 1000
    def, err := c.getDefine(ctx)
    if err != nil {
        ctx.AbortWithError(404, err)
        return
    }
    dt := ctx.PostForm("dt")
    partition := ctx.PostForm("partition")
    data := ctx.PostForm("data")
    if len(dt) != def.DtLength {
        ctx.AbortWithError(404, fmt.Errorf("dt length not match define"))
        return
    }
    var rows []Row
    if err := json.Unmarshal([]byte(data), rows); err != nil {
        ctx.AbortWithError(404, err)
        return
    }
    c.mu.Lock()
    defer c.mu.Unlock()
    store, err := c.stores.GetOrCreate(*def, Partition{partition: partition, dt: dt})
    if err := json.Unmarshal([]byte(data), rows); err != nil {
        ctx.AbortWithError(404, err)
        return
    }
    if !store.IsActive() {
        if err := store.Open(); err != nil {
            ctx.AbortWithError(500, err)
            return
        }
    }
    if err := store.Write(rows); err != nil {
        ctx.AbortWithError(500, err)
        return
    }
    delta := t - time.Now().UnixNano()/1000/1000
    ctx.JSON(200, gin.H{
        "cost_ms": delta,
        "status":  "ok",
    })
}

func (c *Controller) readGetHandler(ctx *gin.Context) {
    t := time.Now().UnixNano() / 1000 / 1000
    def, err := c.getDefine(ctx)
    if err != nil {
        ctx.AbortWithError(404, err)
        return
    }
    key := ctx.Param("Key")
    dtPrefix := ctx.Query("dt")
    partition := ctx.Query("partition")
    c.mu.RLock()
    defer c.mu.RUnlock()
    ss := c.stores.FilterByTable(def.Appid, def.TableName).FilterByPartition(partition, dtPrefix)
    var ret []RowData
    for _, s := range ss {
        if !s.IsActive() {
            if err := s.Open(); err != nil {
                ctx.AbortWithError(500, err)
                return
            }
        }
        value, err := s.Query(key)
        if err != nil {
            ctx.AbortWithError(500, err)
            return
        }
        rd := RowData{
            Partition: s.GetPartition().partition,
            Dt:        s.GetPartition().dt,
            Result:    map[string]string{key: value},
        }
        ret = append(ret, rd)
    }
    delta := t - time.Now().UnixNano()/1000/1000
    ctx.JSON(200, gin.H{
        "cost_ms": delta,
        "data":    ret,
    })
}

func (c *Controller) readPostHandler(ctx *gin.Context) {
    t := time.Now().UnixNano() / 1000 / 1000
    def, err := c.getDefine(ctx)
    if err != nil {
        ctx.AbortWithError(404, err)
        return
    }
    key_prefix := ctx.PostForm("key_prefix")
    column_prefix := ctx.PostForm("column_prefix")
    column_equal := ctx.PostForm("column_equal")
    key_only := ctx.PostForm("key_only") != ""
    dtPrefix := ctx.PostForm("dt")
    partition := ctx.PostForm("partition")
    
    cond := QueryCondition{
        keyPrefix: key_prefix,
    }
    var idx map[string]string
    switch {
    case column_prefix != "" && column_equal == "":
        err := json.Unmarshal([]byte(column_prefix), idx)
        if err != nil {
            ctx.AbortWithError(404, err)
            return
        }
        cond.index = idx
        cond.indexPrefixQuery = true
    case column_equal != "" && column_prefix == "":
        err := json.Unmarshal([]byte(column_equal), idx)
        if err != nil {
            ctx.AbortWithError(404, err)
            return
        }
        cond.index = idx
        cond.indexPrefixQuery = false
    default:
        ctx.AbortWithError(403, fmt.Errorf("column_prefix column_equal conflict"))
        return
    }
    c.mu.RLock()
    defer c.mu.RUnlock()
    ss := c.stores.FilterByTable(def.Appid, def.TableName).FilterByPartition(partition, dtPrefix)
    var ret []RowData
    for _, s := range ss {
        if !s.IsActive() {
            if err := s.Open(); err != nil {
                ctx.AbortWithError(500, err)
                return
            }
        }
        if key_only {
            keys, err := s.QueryKeys(cond)
            if err != nil {
                ctx.AbortWithError(500, err)
                return
            }
            rd := RowData{
                Partition: s.GetPartition().partition,
                Dt:        s.GetPartition().dt,
                Keys:      keys,
            }
            ret = append(ret, rd)
        } else {
            values, err := s.QueryValues(cond)
            if err != nil {
                ctx.AbortWithError(500, err)
                return
            }
            rd := RowData{
                Partition: s.GetPartition().partition,
                Dt:        s.GetPartition().dt,
                Result:    values,
            }
            ret = append(ret, rd)
        }
    }
    delta := t - time.Now().UnixNano()/1000/1000
    ctx.JSON(200, gin.H{
        "cost_ms": delta,
        "data":    ret,
    })
}
